package cn.itcast.dmp.etl

import cn.itcast.dmp.bean.IPUtils._
import cn.itcast.dmp.bean.{IPUtils, IpRegion}
import cn.itcast.dmp.config.AppConfigHelper
import cn.itcast.dmp.process.Processor
import com.maxmind.geoip.LookupService
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{DoubleType, StringType, StructType}
import org.apache.spark.sql.{DataFrame, Row}
import org.lionsoul.ip2region.{DbConfig, DbSearcher}


//对原数据中的IP地址进行转换处理
object IPProcessor  extends Processor{

  /**
  1. 遍历DataFrame，提取IP地址值，通过工具类转换出经纬度省市，并拼接到Row上；
		2. 在原有Schema的基础之上添加新的字段信息，得到新的 Schema
		3. 使用SparkSession，将新的RDD加上新的 Schema，获取生成新的 DataFrame
    */
  override def processData(dataFrame: DataFrame): DataFrame = {
    // 1.针对每个分区数据进行操作，主要：提取IP地址，转换IpRegion
    val newRDD: RDD[Row] =dataFrame.rdd.mapPartitions{
      rows =>
        val dbSearcher = new DbSearcher(new DbConfig(),AppConfigHelper.IPS_DATA_REGION_PATH)
        val service = new LookupService(AppConfigHelper.IPS_DATA_GEO_PATH, LookupService.GEOIP_MEMORY_CACHE)
        // 遍历每个分区的数据，进行转换
        rows.map{row =>
          // 提取IP地址
          val ipValue: String = row.getAs[String]("ip")
          // 通过工具类转换出经纬度省市
          // val ipRegion: IpRegion = IPUtils.convertIp2Region(ipValue)
          val ipRegion: IpRegion = IPUtils.convertIp2Region(ipValue, dbSearcher, service)

          // 将解析获取的省份、城市、经纬度及GeoHash字段值，拼接到原始Row上
          // 对于Seq序列来说， 其中函数 :+ 表示追加到后面，函数 +: 表示追加到前面
          val newSeq: Seq[Any] = row.toSeq :+
            ipRegion.province :+
            ipRegion.city :+
            ipRegion.longitude :+
            ipRegion.latitude :+
            ipRegion.geoHash

          // 返回最新Row
          Row.fromSeq(newSeq)
        }
    }

    // 2. 在原有Schema的基础之上添加新的字段信息，得到新的 Schema
    //只有dataFrame.schema.add 可以有这个方法 dataSet 可以添加schema信息
    val newSchema: StructType = dataFrame.schema
      .add("province", StringType, nullable = true)
      .add("city", StringType, nullable = true)
      .add("longitude", DoubleType, nullable = true)
      .add("latitude", DoubleType, nullable = true)
      .add("geoHash", StringType, nullable = true)


    // 3. 使用SparkSession，将新的RDD加上新的 Schema，获取生成新的 DataFrame
    dataFrame.sparkSession.createDataFrame(newRDD, newSchema)
  }
}
